package tk.xboot.flink.cep.mc;

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.IngestionTimeExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import tk.xboot.flink.cep.mc.events.MonitoringEvent;
import tk.xboot.flink.cep.mc.events.TemperatureAlert;
import tk.xboot.flink.cep.mc.events.TemperatureEvent;
import tk.xboot.flink.cep.mc.events.TemperatureWarning;
import tk.xboot.flink.cep.mc.sources.MonitoringEventSource;

import java.util.List;
import java.util.Map;


public class CEPMonitoring {

    private static final double TEMPERATURE_THRESHOLD = 100;

    private static final double TEMPERATURE_RATIO = 0.5;

    private static final int MAX_RACK_ID = 10;

    private static final long PAUSE = 10;

    private static final double TEMP_STD = 20;

    private static final double TEMP_MEAN = 80;

    private static final double POWER_STD = 10;

    private static final double POWER_MEAN = 100;


    /**
     * CEP Demo from github.
     * https://github.com/tillrohrmann/cep-monitoring/
     *
     *
     * @param args
     */

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // Use ingestion time => TimeCharacteristic == EventTime + IngestionTimeExtractor
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStream<MonitoringEvent> inputEventStream = env.addSource(new MonitoringEventSource(
                MAX_RACK_ID,
                PAUSE,
                TEMPERATURE_RATIO,
                POWER_STD,
                POWER_MEAN,
                TEMP_STD,
                TEMP_MEAN
        )).assignTimestampsAndWatermarks(new IngestionTimeExtractor<>());


        // Equals or greater than TEMPERATURE_THRESHOLD in 10 seconds.
        Pattern<MonitoringEvent, ?> warningPattern = Pattern.<MonitoringEvent>begin("first")
                .subtype(TemperatureEvent.class)
                .where(new IterativeCondition<TemperatureEvent>() {
                    @Override
                    public boolean filter(TemperatureEvent value, Context<TemperatureEvent> ctx) throws Exception {
                        return value.getTemperature() >= TEMPERATURE_THRESHOLD;
                    }
                })
                .next("second")
                .subtype(TemperatureEvent.class)
                .where(new IterativeCondition<TemperatureEvent>() {
                    @Override
                    public boolean filter(TemperatureEvent value, Context<TemperatureEvent> ctx) throws Exception {
                        return value.getTemperature() >= TEMPERATURE_THRESHOLD;
                    }
                })
                .within(Time.seconds(10));

        // Create a pattern stream from our warning pattern
        PatternStream<MonitoringEvent> tempPatternStream = CEP.pattern(
                inputEventStream.keyBy("rackId"),
                warningPattern
        );

        DataStream<TemperatureWarning> warnings = tempPatternStream.select(
                (Map<String, List<MonitoringEvent>> pattern) -> {
                    TemperatureEvent first = (TemperatureEvent) pattern.get("first").get(0);
                    TemperatureEvent second = (TemperatureEvent) pattern.get("second").get(0);
                    return new TemperatureWarning(first.getRackId(),
                            (first.getTemperature() + second.getTemperature()) / 2);
                }
        );

        // Alert pattern: Two consecutive temperature warnings appearing within a time interval of 20 seconds
        Pattern<TemperatureWarning, ?> alertPattern = Pattern.<TemperatureWarning>begin("first")
                .next("second")
                .within(Time.seconds(20));

        // Create a pattern stream from our alert pattern
        PatternStream<TemperatureWarning> alertPatternStream = CEP.pattern(
                warnings.keyBy("rackId"),
                alertPattern
        );

      /*  DataStream<TemperatureAlert> alerts = alertPatternStream.flatSelect(
                new PatternFlatSelectFunction<TemperatureWarning, TemperatureAlert>() {
                    @Override
                    public void flatSelect(Map<String, List<TemperatureWarning>> pattern,
                                           Collector<TemperatureAlert> out) throws Exception {
                        TemperatureWarning first = pattern.get("first").get(0);
                        TemperatureWarning second = pattern.get("second").get(0);
                        if (first.getAverageTemperature() < second.getAverageTemperature()) {
                            out.collect(new TemperatureAlert(first.getRackId()));
                        }
                    }
                }
        );*/


        DataStream<TemperatureAlert> alerts = alertPatternStream
                .flatSelect(
                        (Map<String, List<TemperatureWarning>> pattern, Collector<TemperatureAlert> out) -> {
                            TemperatureWarning first = pattern.get("first").get(0);
                            TemperatureWarning second = pattern.get("second").get(0);
                            if (first.getAverageTemperature() < second.getAverageTemperature()) {
                                out.collect(new TemperatureAlert(first.getRackId()));
                            }
                        },
                        TypeInformation.of(TemperatureAlert.class));


        // Print the warning and alert events to stdout
        warnings.print();
        alerts.print();

        try {
            env.execute("CEP monitoring job");
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
